package com.kld.data.scheduler;

import com.kld.common.utils.DateUtils;
import com.kld.common.utils.StringUtils;
import com.kld.data.domain.po.SysDruid;
import com.kld.data.domain.po.SysTrans;
import com.kld.data.mapper.SysTransMapper;
import com.kld.data.service.ISysDruidService;
import com.kld.data.service.KettleSqlAboutService;
import io.vavr.Tuple4;
import lombok.RequiredArgsConstructor;
import org.pentaho.di.core.KettleEnvironment;
import org.pentaho.di.core.database.DatabaseMeta;
import org.pentaho.di.core.exception.KettleException;
import org.pentaho.di.core.plugins.PluginRegistry;
import org.pentaho.di.core.plugins.StepPluginType;
import org.pentaho.di.trans.Trans;
import org.pentaho.di.trans.TransHopMeta;
import org.pentaho.di.trans.TransMeta;
import org.pentaho.di.trans.step.StepMeta;
import org.pentaho.di.trans.step.StepMetaInterface;
import org.pentaho.di.trans.steps.insertupdate.InsertUpdateMeta;
import org.pentaho.di.trans.steps.rowstoresult.RowsToResultMeta;
import org.pentaho.di.trans.steps.tableinput.TableInputMeta;
import org.pentaho.di.trans.steps.tableoutput.TableOutputMeta;
import org.pentaho.di.trans.steps.userdefinedjavaclass.UserDefinedJavaClassDef;
import org.pentaho.di.trans.steps.userdefinedjavaclass.UserDefinedJavaClassDef.ClassType;
import org.pentaho.di.trans.steps.userdefinedjavaclass.UserDefinedJavaClassMeta;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Service;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Date;
import java.util.List;

@RequiredArgsConstructor
@Service
public class KettleService {

    private static final Logger LOG = LoggerFactory.getLogger(KettleService.class);

    private static final String DBA01_SQL_TEMPLATE = "SELECT JH,RQ,SCSJ,RCQL,RCYL,RCYL1,RCSL,TY,YY,HY,HDL,HCYL,QYB,BZ,BZDM,BZDM1,VRUSERNAME,LRSJ,UPLOADFLAG,IP,XH,HS,JRKC,ZRYY,JRFY,JRPS,QTYY,RCYZJ,GJYY FROM YS_DBA01 " +
        "WHERE RQ >= TO_DATE( '${1}', 'yyyy-mm-dd hh24:mi:ss' ) AND LRSJ >= TO_DATE( '${2}', 'yyyy-mm-dd hh24:mi:ss' ) AND LRSJ <= TO_DATE( '${3}', 'yyyy-mm-dd hh24:mi:ss' )";

    StringBuilder jsonStr;

    private final ISysDruidService druidService;

    private final SysTransMapper sysTransMapper;

    private final KettleSqlAboutService kettleSqlAboutService;

    /**
     * 执行转换任务
     *
     * @throws IOException
     * @throws KettleException
     */
    @Async
    public void excuteTrans(SysTrans systrans) {
        final String tableInputSql = systrans.getTableInputSql(); // 原始的SQL， 中间可能有占位符
        try {
            KettleEnvironment.init();
            TransMeta transMeta = null;
            SysDruid sysDruid = druidService.selectSysDruidById(systrans.getTargetId());
            transMeta = getTransMeta(systrans);
//            if(sysDruid!=null){
//                if("mongodb".equals(sysDruid.getType())){
//                    transMeta = getTransMetaToMongodb(systrans);
//                }else{
//                    transMeta = getTransMeta(systrans);
//                }
//            }else{
//                transMeta = getTransMeta(systrans);
//            }
            Trans trans = new Trans(transMeta);
            trans.prepareExecution(null);
            trans.startThreads();
            trans.waitUntilFinished();
            systrans.setLastExecTime(new Date());
            systrans.setLastExecResult(1);
            int errors = trans.getErrors();
            if (errors > 0) {
                throw new Exception("kettle 内部执行异常");
            }
        } catch (Exception e) {
            LOG.error("数据转换发生异常", e);
            systrans.setLastExecTime(new Date());
            systrans.setLastExecResult(0);
//            insertFailNotice(systrans, e);
        } finally {
            KettleEnvironment.shutdown();
        }
        systrans.setUpdateTime(DateUtils.getNowDate());
        systrans.setTableInputSql(tableInputSql);
        sysTransMapper.updateSysTrans(systrans);
    }

    //
    private void setDba01OrDba02DailySql(SysTrans systrans) {
        final String maxLrsj = kettleSqlAboutService.findDba01OrDba02MaxLrsj(systrans.getTableName());
        final String newSql = systrans.getTableInputSql().replaceAll("\\$\\{1}", maxLrsj);
        systrans.setTableInputSql(newSql);
    }

    //
    private void setSql(SysTrans systrans, Tuple4<String, String, String, String> dateTuple) {
        String finalSql = DBA01_SQL_TEMPLATE.replace("${1}", dateTuple._2())
            .replace("${2}", dateTuple._3())
            .replace("${3}", dateTuple._4());
        systrans.setTableInputSql(finalSql);
    }

//    private void insertFailNotice(SysTrans systrans, Exception e) {
//        SysNotice notice = new SysNotice();
//        notice.setCreateBy("系统后台");
////        notice.setUserNickname("系统后台");
//        notice.setNoticeType("1");
//        notice.setNoticeTitle("数据转换失败通知");
//        SimpleDateFormat format = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
//        notice.setNoticeContent(format.format(new Date()) + " 您名称为“" + systrans.getTaskName() + "”数据转换任务执行失败，原因为：数据转换发生异常，" + e.getMessage());
//        sysNoticeMapper.insertNotice(notice);
//    }


    /**
     * 根据转换任务构造转换对象
     *
     * @param systrans
     * @return
     */
    public TransMeta getTransMeta(SysTrans systrans) throws KettleException {

        TransMeta transMeta = new TransMeta();

        // 设置转化的名称
        transMeta.setName("insert_update");

        // 添加转换的数据库连接
        String sourceId = systrans.getSourceId();
        SysDruid sourceDruid = druidService.selectSysDruidById(sourceId);
        DatabaseMeta sourceMeta = sourceDruid.initDatabaseMeta();
        //设置字符集
        //sourceMeta.addExtraOption("MYSQL", "characterEncoding", "utf-8");

        String targetId = systrans.getTargetId();
        SysDruid tragetDruid = druidService.selectSysDruidById(targetId);
        DatabaseMeta targetMeta = tragetDruid.initDatabaseMeta();

        // registry是给每个步骤生成一个标识id
        PluginRegistry registry = PluginRegistry.getInstance();

        // 第一个表输入步骤(TableInputMeta)
        TableInputMeta tableInput = new TableInputMeta();
        tableInput.setDatabaseMeta(sourceMeta);
        tableInput.setSQL(systrans.getTableInputSql().replace("&gt;", ">").replace("&lt;", "<"));
        String tableInputPluginId = registry.getPluginId(StepPluginType.class, tableInput);
        StepMeta tableInputMetaStep = new StepMeta(tableInputPluginId, "表输入", tableInput);
        transMeta.addStep(tableInputMetaStep);

        StepMeta stepMeta = null;
        String outputType = systrans.getOutputType();
        if ("0".equals(outputType)) {
            // 第二步为表输出
            TableOutputMeta tableOutputMeta = new TableOutputMeta();
            String tableOutputMetaPluginId = registry.getPluginId(StepPluginType.class, tableOutputMeta);
            tableOutputMeta.setDatabaseMeta(targetMeta);
            tableOutputMeta.setSchemaName(targetMeta.getDatabaseName());
            tableOutputMeta.setTableName(systrans.getTableName());
            tableOutputMeta.setCommitSize(10000);
//            tableOutputMeta.setIgnoreErrors(true);
            tableOutputMeta.setUseBatchUpdate(true);
            StepMeta tableOutputStep = new StepMeta(tableOutputMetaPluginId, "table_output", tableOutputMeta);
            transMeta.addStep(tableOutputStep);
            stepMeta = tableOutputStep;
        } else {
            // 第二步为插入与更新
            InsertUpdateMeta insertUpdateMeta = new InsertUpdateMeta();
            String insertUpdateMetaPluginId = registry.getPluginId(StepPluginType.class, insertUpdateMeta);
            insertUpdateMeta.setDatabaseMeta(targetMeta);
            insertUpdateMeta.setTableName(systrans.getTableName());
            // 设置用来查询的关键字
            insertUpdateMeta.setKeyLookup(systrans.getLookupKey().split(","));
            insertUpdateMeta.setKeyCondition(systrans.getConditionKey().split(","));
            insertUpdateMeta.setKeyStream(systrans.getStreamKey().split(","));
            insertUpdateMeta.setKeyStream2(new String[]{"", "", ""});
            // 设置要更新的字段
            String[] updatelookup = systrans.getLookupColums().split(",");
            String[] updateStream = systrans.getStreamColums().split(",");
            String updateExceptions = systrans.getExceptionColums();
            Boolean[] updateOrNot = new Boolean[updateStream.length];
            int keyNum = systrans.getLookupKey().split(",").length;
            for (int i = 0; i < updateOrNot.length; i++) {
                if (i < keyNum) {
                    updateOrNot[i] = false;
                } else {
                    if(StringUtils.isNotBlank(updateExceptions)){
                        if(updateExceptions.indexOf(updateStream[i]) > -1){
                            updateOrNot[i] = false;
                        }else{
                            updateOrNot[i] = true;
                        }
                    }else{
                        updateOrNot[i] = true;
                    }
                }
            }
            insertUpdateMeta.setUpdateLookup(updatelookup);
            insertUpdateMeta.setUpdateStream(updateStream);
            insertUpdateMeta.setUpdate(updateOrNot);
            StepMeta insertUpdateStep = new StepMeta(insertUpdateMetaPluginId, "insert_update", insertUpdateMeta);
            transMeta.addStep(insertUpdateStep);
            stepMeta = insertUpdateStep;
        }

        // 复制记录到结果
        RowsToResultMeta rowsToResultMeta = new RowsToResultMeta();
        String rowsToResultMetaPluginId = registry.getPluginId(
            StepPluginType.class, rowsToResultMeta);
        // 添加步骤到转换中
        StepMeta rowsToResultStepMeta = new StepMeta(rowsToResultMetaPluginId,
            "rowsToResult", (StepMetaInterface) rowsToResultMeta);
        transMeta.addStep(rowsToResultStepMeta);

        // 是否调用Java
        if (systrans.getIsNeedJava()) {

            // 中间步骤，调用Java代码
            UserDefinedJavaClassMeta javaClassMeta = new UserDefinedJavaClassMeta();
            String javaClassMetaPluginId = registry.getPluginId(StepPluginType.class, javaClassMeta);
            UserDefinedJavaClassDef javaDef = new UserDefinedJavaClassDef(ClassType.TRANSFORM_CLASS, "Processor", systrans.getJavaCode());
            List<UserDefinedJavaClassDef> defList = new ArrayList<UserDefinedJavaClassDef>();
            defList.add(javaDef);
            javaClassMeta.replaceDefinitions(defList);

            // 添加步骤到转换中
            StepMeta javaClassStep = new StepMeta(javaClassMetaPluginId, "java_code", javaClassMeta);
            javaClassStep.setDraw(true);
            javaClassStep.setLocation(250, 100);
            transMeta.addStep(javaClassStep);
            transMeta.addTransHop(new TransHopMeta(tableInputMetaStep, javaClassStep));
            transMeta.addTransHop(new TransHopMeta(javaClassStep, stepMeta));

        } else {
            transMeta.addTransHop(new TransHopMeta(tableInputMetaStep, stepMeta));
        }

        return transMeta;
    }


//    /**
//     * 根据转换任务构造转换对象-mongodb
//     *
//     * @param systrans
//     * @return
//     */
//    public TransMeta getTransMetaToMongodb(SysTrans systrans) throws KettleException {
//
//        TransMeta transMeta = new TransMeta();
//
//        // 设置转化的名称
//        transMeta.setName("insert_update");
//
//        // 添加转换的数据库连接
//        String sourceId = systrans.getSourceId();
//        SysDruid sourceDruid = druidService.selectSysDruidById(sourceId);
//        DatabaseMeta sourceMeta = sourceDruid.initDatabaseMeta();
//        //设置字符集
//        //sourceMeta.addExtraOption("MYSQL", "characterEncoding", "utf-8");
//
//        String targetId = systrans.getTargetId();
//        SysDruid tragetDruid = druidService.selectSysDruidById(targetId);
//
//        // registry是给每个步骤生成一个标识id
//        PluginRegistry registry = PluginRegistry.getInstance();
//
//        // 第一个表输入步骤(TableInputMeta)
//        TableInputMeta tableInput = new TableInputMeta();
//        tableInput.setDatabaseMeta(sourceMeta);
//        tableInput.setSQL(systrans.getTableInputSql().replace("&gt;", ">").replace("&lt;", "<"));
//        String tableInputPluginId = registry.getPluginId(StepPluginType.class, tableInput);
//        StepMeta tableInputMetaStep = new StepMeta(tableInputPluginId, "表输入", tableInput);
//        transMeta.addStep(tableInputMetaStep);
//
//        StepMeta stepMeta = null;
//        String outputType = systrans.getOutputType();
//        /* 第二步为mongdb输出 */
//        MongoDbOutputMeta mongoDbOutputMeta = new MongoDbOutputMeta();
//        String mongoOutputMetaPluginId = registry.getPluginId(StepPluginType.class, mongoDbOutputMeta);
//        //设置认证
//        mongoDbOutputMeta.setCollection(systrans.getTableName());
//        mongoDbOutputMeta.setPort(tragetDruid.getPort());
//        mongoDbOutputMeta.setHostnames(tragetDruid.getHost());
//        mongoDbOutputMeta.setDbName(tragetDruid.getDb());
//        mongoDbOutputMeta.setAuthenticationMechanism("SCRAM-SHA-1");
//        mongoDbOutputMeta.setAuthenticationUser(tragetDruid.getUserName());
//        mongoDbOutputMeta.setAuthenticationPassword(tragetDruid.getPassWord());
//        mongoDbOutputMeta.setBatchInsertSize(String.valueOf(10000));
//        mongoDbOutputMeta.setTruncate(true);
////        mongoDbOutputMeta.setUpsert(true);
////        mongoDbOutputMeta.setUpdate(true);
////        mongoDbOutputMeta.setModifierUpdate(true);
////        mongoDbOutputMeta.setMulti(true);
//        // 字段列表
//        String[] fileds = systrans.getStreamColums().split(",");
//        List<MongoDbOutputMeta.MongoField> fieldList = new ArrayList();
//        for (int i = 0; i < fileds.length; i++) {
//            MongoDbOutputMeta.MongoField field = new MongoDbOutputMeta.MongoField();
//            field.m_incomingFieldName = fileds[i];
//            field.m_useIncomingFieldNameAsMongoFieldName = true;
//            fieldList.add(field);
//        }
//        mongoDbOutputMeta.setMongoFields(fieldList);
//        StepMeta tableOutputStep = new StepMeta(mongoOutputMetaPluginId, "mongodb_output", mongoDbOutputMeta);
//        transMeta.addStep(tableOutputStep);
//        stepMeta = tableOutputStep;
//
//        // 复制记录到结果
//        RowsToResultMeta rowsToResultMeta = new RowsToResultMeta();
//        String rowsToResultMetaPluginId = registry.getPluginId(
//                StepPluginType.class, rowsToResultMeta);
//        // 添加步骤到转换中
//        StepMeta rowsToResultStepMeta = new StepMeta(rowsToResultMetaPluginId,
//                "rowsToResult", (StepMetaInterface) rowsToResultMeta);
//        transMeta.addStep(rowsToResultStepMeta);
//
//        // 是否调用Java
//        if (systrans.getIsNeedJava()) {
//
//            // 中间步骤，调用Java代码
//            UserDefinedJavaClassMeta javaClassMeta = new UserDefinedJavaClassMeta();
//            String javaClassMetaPluginId = registry.getPluginId(StepPluginType.class, javaClassMeta);
//            UserDefinedJavaClassDef javaDef = new UserDefinedJavaClassDef(ClassType.TRANSFORM_CLASS, "Processor", systrans.getJavaCode());
//            List<UserDefinedJavaClassDef> defList = new ArrayList<UserDefinedJavaClassDef>();
//            defList.add(javaDef);
//            javaClassMeta.replaceDefinitions(defList);
//
//            // 添加步骤到转换中
//            StepMeta javaClassStep = new StepMeta(javaClassMetaPluginId, "java_code", javaClassMeta);
//            javaClassStep.setDraw(true);
//            javaClassStep.setLocation(250, 100);
//            transMeta.addStep(javaClassStep);
//            transMeta.addTransHop(new TransHopMeta(tableInputMetaStep, javaClassStep));
//            transMeta.addTransHop(new TransHopMeta(javaClassStep, stepMeta));
//
//        } else {
//            transMeta.addTransHop(new TransHopMeta(tableInputMetaStep, stepMeta));
//        }
//
//        return transMeta;
//    }


}
